package mq.job;

import common.MyDisReentrantLock;
import lombok.extern.slf4j.Slf4j;
import mq.domain.MqMessageAbnormal;
import mq.service.IMqMessageAbnormalService;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.springframework.scheduling.quartz.QuartzJobBean;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @description: MQ失败消息重新发送
 * @author: zk
 * @since: 2022-08-21 09:26:41
 */
@Slf4j
public class FailedMessageResendJob extends QuartzJobBean {
    private final String MQ_FAILED_MESSAGE_RESEND_LOCK = "MQ_FAILED_MESSAGE_RESEND_LOCK";
    @Resource
    private MyDisReentrantLock myDisReentrantLock;
    @Resource
    private IMqMessageAbnormalService mqMessageAbnormalService;


    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        log.info("MQ失败消息重新发送定时任务 begin");
        boolean lock = myDisReentrantLock.lock(FailedMessageResendJob.class, MQ_FAILED_MESSAGE_RESEND_LOCK, 7200);
        if (!lock) {
            log.info("MQ失败消息重新发送定时任务,已经有任务在处理 end");
            return;
        }
        try {
            try {
                mqMessageAbnormalService.resendMessage(MqMessageAbnormal.TYPE_EXCHANGE);
            } catch (Exception e) {
                log.error("MQ失败消息重新发送-TYPE_EXCHANGE 执行失败", e);
            }
            TimeUnit.SECONDS.sleep(5);
            try {
                mqMessageAbnormalService.resendMessage(MqMessageAbnormal.TYPE_QUEUE);
            } catch (Exception e) {
                log.error("MQ失败消息重新发送-TYPE_QUEUE 执行失败", e);
            }
        } catch (Exception e) {
            log.error("MQ失败消息重新发送执行失败", e);
        } finally {
            myDisReentrantLock.unLock(FailedMessageResendJob.class, MQ_FAILED_MESSAGE_RESEND_LOCK);
            log.info("MQ失败消息重新发送定时任务 end");
        }
    }
}
